package app.dws;

import app.ods.Constant;
import app.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.types.Row;

/**
 * DWS层：用户家庭特征标签汇总计算
 * 核心逻辑：按用户维度汇总性别得分、年龄得分，生成宝宝性别、宝宝年龄标签
 */
public class UserFamilyTagAggregation {
    // 自定义聚合函数：计算各年龄分段的总得分
    public static class AgeStageScoreAgg extends AggregateFunction<Double, Double> {
        @Override
        public Double createAccumulator() {
            return 0.0;
        }

        @Override
        public Double getValue(Double accumulator) {
            return accumulator;
        }

        // 累加计算：年龄得分 = 年龄权重 × 行为权重 × 时间衰减系数
        public void accumulate(Double accumulator, Double ageWeight, Double behaviorWeight, Double timeDecay) {
            if (ageWeight != null && behaviorWeight != null && timeDecay != null) {
                accumulator += ageWeight * behaviorWeight * timeDecay;
            }
        }
    }

    public static void main(String[] args) throws Exception {
        // 1. 初始化Flink执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        EnvironmentSettings tableEnvSettings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, tableEnvSettings);

        // 2. 注册自定义聚合函数
        tableEnv.createTemporaryFunction("AgeStageScoreAgg", AgeStageScoreAgg.class);

        // 3. 注册Kafka DWD层数据源表（已扩充权重的数据）
        tableEnv.executeSql("CREATE TABLE dwd_user_behavior_enriched (" +
                "  user_id STRING COMMENT '用户ID'," +
                "  goods_id STRING COMMENT '商品ID'," +
                "  behavior_type STRING COMMENT '行为类型'," +
                "  behavior_time BIGINT COMMENT '行为时间（毫秒）'," +
                "  search_keyword STRING COMMENT '搜索关键词'," +
                "  goods_category STRING COMMENT '商品类目'," +
                "  gender_weight DOUBLE COMMENT '商品性别权重'," +
                "  gender_direction STRING COMMENT '商品性别指向（female/male/neutral/none）'," +
                "  age_stage STRING COMMENT '商品关联年龄分段'," +
                "  age_weight DOUBLE COMMENT '商品年龄权重'," +
                "  behavior_weight DOUBLE COMMENT '行为权重'," +
                "  time_decay DOUBLE COMMENT '时间衰减系数'," +
                "  ts TIMESTAMP_LTZ(3) COMMENT '行为时间戳'," +
                "  WATERMARK FOR ts AS ts - INTERVAL '5' SECOND" +
                ") "+ MyKafkaUtil.getKafkaDDL("dwd_user_behavior_enriched", "dwd_user_behavior_enriched_1565"));

        // 4. 注册Kafka DWS层输出表（最终用户标签）
        tableEnv.executeSql("CREATE TABLE dws_user_family_tag (" +
                "  user_id STRING COMMENT '用户ID'," +
                "  baby_gender STRING COMMENT '宝宝性别（男/女/未识别）'," +
                "  baby_age_stage STRING COMMENT '宝宝年龄分段'," +
                "  calculate_time TIMESTAMP_LTZ(3) COMMENT '标签计算时间'," +
                "  PRIMARY KEY (user_id) NOT ENFORCED" +
                ") "+ MyKafkaUtil
                .getKafkaDDL("dws_user_family_tag", "dws_user_family_tag_1565"));

        // 5. 核心SQL：标签汇总计算（DWS层核心逻辑）
        String dwsTagSql = "INSERT INTO dws_user_family_tag " +
                "SELECT " +
                "  user_id," +
                "  -- 宝宝性别判定：女宝得分 vs 男宝得分" +
                "  CASE " +
                "    WHEN female_score > male_score THEN '女'" +
                "    WHEN female_score < male_score THEN '男'" +
                "    ELSE '未识别' " +
                "  END AS baby_gender," +
                "  -- 宝宝年龄分段判定：得分最高且≥其他两倍，否则取最新行为对应的分段" +
                "  CASE " +
                "    WHEN early_pregnancy_score >= 2 * GREATEST(mid_pregnancy_score, late_pregnancy_score, o3m_score, o6m_score, o12m_score, o24m_score, o36m_score, o36y_score, o12y_score) THEN '" + Constant.AGE_STAGE_EARLY_PREGNANCY + "'" +
                "    WHEN mid_pregnancy_score >= 2 * GREATEST(early_pregnancy_score, late_pregnancy_score, o3m_score, o6m_score, o12m_score, o24m_score, o36m_score, o36y_score, o12y_score) THEN '" + Constant.AGE_STAGE_MID_PREGNANCY + "'" +
                "    WHEN late_pregnancy_score >= 2 * GREATEST(early_pregnancy_score, mid_pregnancy_score, o3m_score, o6m_score, o12m_score, o24m_score, o36m_score, o36y_score, o12y_score) THEN '" + Constant.AGE_STAGE_LATE_PREGNANCY + "'" +
                "    WHEN o3m_score >= 2 * GREATEST(early_pregnancy_score, mid_pregnancy_score, late_pregnancy_score, o6m_score, o12m_score, o24m_score, o36m_score, o36y_score, o12y_score) THEN '" + Constant.AGE_STAGE_0_3M + "'" +
                "    WHEN o6m_score >= 2 * GREATEST(early_pregnancy_score, mid_pregnancy_score, late_pregnancy_score, o3m_score, o12m_score, o24m_score, o36m_score, o36y_score, o12y_score) THEN '" + Constant.AGE_STAGE_3_6M + "'" +
                "    WHEN o12m_score >= 2 * GREATEST(early_pregnancy_score, mid_pregnancy_score, late_pregnancy_score, o3m_score, o6m_score, o24m_score, o36m_score, o36y_score, o12y_score) THEN '" + Constant.AGE_STAGE_6_12M + "'" +
                "    WHEN o24m_score >= 2 * GREATEST(early_pregnancy_score, mid_pregnancy_score, late_pregnancy_score, o3m_score, o6m_score, o12m_score, o36m_score, o36y_score, o12y_score) THEN '" + Constant.AGE_STAGE_12_24M + "'" +
                "    WHEN o36m_score >= 2 * GREATEST(early_pregnancy_score, mid_pregnancy_score, late_pregnancy_score, o3m_score, o6m_score, o12m_score, o24m_score, o36y_score, o12y_score) THEN '" + Constant.AGE_STAGE_24_36M + "'" +
                "    WHEN o36y_score >= 2 * GREATEST(early_pregnancy_score, mid_pregnancy_score, late_pregnancy_score, o3m_score, o6m_score, o12m_score, o24m_score, o36m_score, o12y_score) THEN '" + Constant.AGE_STAGE_3_6Y + "'" +
                "    WHEN o12y_score >= 2 * GREATEST(early_pregnancy_score, mid_pregnancy_score, late_pregnancy_score, o3m_score, o6m_score, o12m_score, o24m_score, o36m_score, o36y_score) THEN '" + Constant.AGE_STAGE_6_12Y + "'" +
                "    ELSE latest_age_stage " + // 得分相近时，取最新行为的年龄分段
                "  END AS baby_age_stage," +
                "  CURRENT_TIMESTAMP() AS calculate_time " +
                "FROM (" +
                "  -- 子查询1：按用户汇总各维度得分" +
                "  SELECT " +
                "    user_id," +
                "    -- 女宝得分：性别指向为female的商品得分总和" +
                "    SUM(CASE WHEN gender_direction = 'female' THEN gender_weight * behavior_weight * time_decay ELSE 0 END) AS female_score," +
                "    -- 男宝得分：性别指向为male的商品得分总和" +
                "    SUM(CASE WHEN gender_direction = 'male' THEN gender_weight * behavior_weight * time_decay ELSE 0 END) AS male_score," +
                "    -- 各年龄分段得分（使用自定义聚合函数）" +
                "    AgeStageScoreAgg(age_weight, behavior_weight, time_decay) FILTER (WHERE age_stage = '" + Constant.AGE_STAGE_EARLY_PREGNANCY + "') AS early_pregnancy_score," +
                "    AgeStageScoreAgg(age_weight, behavior_weight, time_decay) FILTER (WHERE age_stage = '" + Constant.AGE_STAGE_MID_PREGNANCY + "') AS mid_pregnancy_score," +
                "    AgeStageScoreAgg(age_weight, behavior_weight, time_decay) FILTER (WHERE age_stage = '" + Constant.AGE_STAGE_LATE_PREGNANCY + "') AS late_pregnancy_score," +
                "    AgeStageScoreAgg(age_weight, behavior_weight, time_decay) FILTER (WHERE age_stage = '" + Constant.AGE_STAGE_0_3M + "') AS o3m_score," +
                "    AgeStageScoreAgg(age_weight, behavior_weight, time_decay) FILTER (WHERE age_stage = '" + Constant.AGE_STAGE_3_6M + "') AS o6m_score," +
                "    AgeStageScoreAgg(age_weight, behavior_weight, time_decay) FILTER (WHERE age_stage = '" + Constant.AGE_STAGE_6_12M + "') AS o12m_score," +
                "    AgeStageScoreAgg(age_weight, behavior_weight, time_decay) FILTER (WHERE age_stage = '" + Constant.AGE_STAGE_12_24M + "') AS o24m_score," +
                "    AgeStageScoreAgg(age_weight, behavior_weight, time_decay) FILTER (WHERE age_stage = '" + Constant.AGE_STAGE_24_36M + "') AS o36m_score," +
                "    AgeStageScoreAgg(age_weight, behavior_weight, time_decay) FILTER (WHERE age_stage = '" + Constant.AGE_STAGE_3_6Y + "') AS o36y_score," +
                "    AgeStageScoreAgg(age_weight, behavior_weight, time_decay) FILTER (WHERE age_stage = '" + Constant.AGE_STAGE_6_12Y + "') AS o12y_score," +
                "    -- 最新行为对应的年龄分段" +
                "    LAST_VALUE(age_stage) OVER (PARTITION BY user_id ORDER BY behavior_time DESC ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS latest_age_stage " +
                "  FROM dwd_user_behavior_enriched " +
                "  -- 按用户分组，基于滚动窗口（1天）更新标签（符合每日更新要求）" +
                "  GROUP BY user_id, TUMBLE(ts, INTERVAL '1' DAY)" +
                ") t1)";

        // 6. 执行DWS层标签计算任务
        tableEnv.executeSql(dwsTagSql);
        System.out.println("DWS层用户标签计算任务启动成功");
        env.execute("UserFamilyTagAggregationTask");
    }
}